package com.bw.sparkstreaming.job4

import org.apache.log4j.{Level, Logger}
import org.apache.spark.SparkConf
import org.apache.spark.streaming.{Seconds, StreamingContext}


/**
  * 实现单词计数的效果
  *
  * hadoop,hadoop,hadoop
  * (hadoop,3)
  *
  *
  * hadoop
  * (hadoop,4) -> (hadoop,1)
  */
object UpdateStateByKeyTest {
  Logger.getLogger("bw").setLevel(Level.ERROR)

  def main(args: Array[String]): Unit = {
    //程序入口
    val conf = new SparkConf().setMaster("local[2]").setAppName(this.getClass.getSimpleName)
    val ssc = new StreamingContext(conf,Seconds(5))

    ssc.checkpoint("E:\\pyworksapce\\day05")


    //输入
    val dataDStream = ssc.socketTextStream("192.168.192.130",9999)

    //数据处理
    val wDStream = dataDStream.flatMap(_.split(",")) // hive,hadoop,hadoop,hbase,flink......
    val wordDStram = wDStream.map((_,1))


    /**
      *
      *  updateFunc:(Seq[v],Option[S]) => Option[S]
      *  参数二：Seq[V]
      *  hive,1
      *  hive,1
      *  hive,1
      *  hive,1
      *  分组：
      *  {hive,(1,1,1,1)} values=> (1,1,1,1)
      *
      *  参数二：Option[S]
      *  当前的这个key的上一次的状态
      *
      *
      *  返回值：Option[S] 当前key出现的次数。
      *
      *  结论：将当前的这个key的状态，和上一次出现的状态加起来作为返回值就行了。

      */

     val result = wordDStram.updateStateByKey((values:Seq[Int],state:Option[Int]) => {
      val currentCount = values.sum
      val lastCount = state.getOrElse(0)
      Some(currentCount+lastCount)
    })

    //输出
    result.print()


    //启动程序
    ssc.start()
    ssc.awaitTermination()
    ssc.stop()
  }

}
